/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.zx.learn.flink.official.frauddetection;

import com.zx.learn.flink.official.frauddetection.entity.Alert;
import com.zx.learn.flink.official.frauddetection.entity.Transaction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Skeleton code for implementing a fraud detector.
 * 定义了欺诈交易检测的业务逻辑
 * <p>
 * 报警程序，对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
 */
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {

    private static final long serialVersionUID = 1L;

    //使用一个 boolean 型的标记状态来表示是否刚处理过一个小额交易。
    //可以使用如 Map 这样的数据结构来保存每一个账户的状态但是常规的类成员变量是无法做到容错处理的，当任务失败重启后，
    // 之前的状态信息将会丢失。 这样的话，如果程序曾出现过失败重启的情况，将会漏掉一些欺诈报警。
    //为了应对这个问题，Flink 提供了一套支持容错状态的原语，这些原语几乎与常规成员变量一样易于使用。

    //ValueState 是一个包装类，类似于 Java 标准库里边的 AtomicReference 和 AtomicLong。 它提供了三个用于交互的方法。update 用于更新状态，value 用于获取状态值，还有
    // clear 用于清空状态。 如果一个 key 还没有状态，例如当程序刚启动或者调用过 ValueState#clear 方法时，ValueState#value 将会返回 null。 如果需要更新状态，需要调用
    // ValueState#update 方法，直接更改 ValueState#value 的返回值可能不会被系统识别。 容错处理将在 Flink 后台自动管理，你可以像与常规变量那样与状态变量进行交互。
    private transient ValueState<Boolean> flagState;

    private static final double SMALL_AMOUNT = 1.00;
    private static final double LARGE_AMOUNT = 500.00;
    private static final long ONE_MINUTE = 60 * 1000;

    /**
     * 该方法在每个交易事件上被调用，会对每笔交易发出警报
     */
    @Override
    public void processElement(
            Transaction transaction,
            Context context,
            Collector<Alert> collector) throws Exception {

        //默认实现，每一个事件触发报警
        //Alert alert = new Alert();
        //alert.setId(transaction.getAccountId());
        //collector.collect(alert);

        //实现：如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
        // Get the current state for the current key
        Boolean lastTransactionWasSmall = flagState.value();
        // Check if the flag is set
        if (lastTransactionWasSmall != null) {
            if (transaction.getAmount() > LARGE_AMOUNT) {
                // Output an alert downstream
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());
                //只有这种情况才会报警.
                collector.collect(alert);
            }
            // Clean up our state ,在检查之后，不论是什么状态，都需要被清空。 不管是当前交易触发了欺诈报警而造成模式的结束，还是当前交易没有触发报警而造成模式的中断，都需要重新开始新的模式检测。
            flagState.clear();
        }

        if (transaction.getAmount() < SMALL_AMOUNT) {
            // Set the flag to true
            flagState.update(true);
        }

    }

    @Override
    public void open(Configuration parameters) {
        //Flink 中最基础的状态类型是 ValueState，这是一种能够为被其封装的变量添加容错能力的类型。 ValueState 是一种 keyed state，也就是说它只能被用于 keyed context
        // 提供的 operator 中，即所有能够紧随 DataStream#keyBy 之后被调用的operator。 一个 operator 中的 keyed state 的作用域默认是属于它所属的 key 的。
        // 这个例子中，key 就是当前正在处理的交易行为所属的信用卡账户（key 传入 keyBy() 函数调用），而 FraudDetector 维护了每个帐户的标记状态。 ValueState 需要使用
        // ValueStateDescriptor 来创建，
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                "flag",
                Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDescriptor);
    }
}
